#pragma once

#include <iostream>
#include <memory>
#include <mutex>
#include <cassert>
#include <unordered_map>
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"


namespace bitmq
{
    using ConsumerCallBack = std::function<void(const std::string&,  const BasicProperties* , const std::string&)>;
    //1. 定义消费者信息结构
    struct Consumer
    {
    public:
        using ptr = std::shared_ptr<Consumer>;
        Consumer(){}
        Consumer(const std::string &ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallBack& cb)
            : tag(ctag)
            , qname(queue_name)
            , callback(cb)
            , auto_ack(ack_flag)
        {
            DLOG("new Consumer: %s", ctag.c_str());
        }
        ~Consumer()
        {
            DLOG("delete Consumer: %s", tag.c_str());
        }

    public:
        std::string tag;   // 1）消费者标识
        std::string qname;  // 2）订阅的队列名称
        ConsumerCallBack callback; // 3）一个消息的处理回调函数
        bool auto_ack;    // 4）是否自动应答标志
    };

    //2. 消费者管理--以队列为单元进行管理-队列消费者管理结构
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;
        QueueConsumer()
        {}

        QueueConsumer(const std::string& qname)
            : _qname(qname)
            , _rr_seq(0)
        {}

        // 1）新增消费者：信道提供的服务是订阅队列消息的时候创建
        Consumer::ptr create(const std::string &ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallBack& cb)
        {
            //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断消费者是否重复
            for (auto& cmp : _consumers)
            {
                if (cmp->tag == ctag)
                    return {};
            }
            //3.没有重复则新增————构造对象
            auto consumer = std::make_shared<Consumer>(ctag, queue_name, ack_flag, cb);
            //4.添加管理后返回对象
            _consumers.push_back(consumer);
            return consumer;
        }

        // 2）删除消费者：取消订阅 / 信道关闭 / 连接关闭 的时候删除
        void remove(const std::string& ctag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->tag == ctag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        // 3）获取消费者：从队列所有的消费者中按序取出一个消费者进行消息的推送
        Consumer::ptr choose()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
                return {};
            //获取轮转下标
            int idx = _rr_seq % _consumers.size();
            ++_rr_seq;
            //返回结果
            return _consumers[idx];
        }
        
        // 4）判断队列消费者是否为空
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }

        // 5）判断指定消费者是否存在
        bool exists(const std::string& ctag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->tag == ctag)
                {
                    return true;
                }
            }
            return false;
        }

        // 6）清理队列所有消费者
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

    private:
        std::string _qname;  // 1）队列名称
        std::vector<Consumer::ptr> _consumers;  // 2）消费者管理结构： vector
        uint64_t _rr_seq;   // 3）轮转序号
        std::mutex _mutex;  // 4）互斥锁：保证线程安全
    };

    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager()
        {}

        // 1）初始化/删除队列的消费者信息结构（创建/删除队列的时候初始化）
        void initQueueConsumer(const std::string& qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it != _qconsumers.end())
                return;

            auto qconsumers = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert({qname, qconsumers});
        }

        // 2）移除指定队列的所有消费者（队列被删除时销毁）：删除消费者的队列管理单元对象
        void destroyQueueConsumer(const std::string& qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it == _qconsumers.end())
                return;
            _qconsumers.erase(it);
        }

        // 3）向指定队列新增消费者（客户端订阅指定队列消息的时候）：新增完成的时候返回消费者对象
        Consumer::ptr create(const std::string &ctag, const std::string& queue_name, bool ack_flag, const ConsumerCallBack& cb)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //获取消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    DLOG("没有找到队列%s的消息管理句柄", queue_name.c_str());
                    return {};
                }
                qcp = it->second;
            }
            //通过句柄完成创建
            return qcp->create(ctag, queue_name, ack_flag, cb);
        }

        // 4）从指定队列移除消费者（客户端取消订阅的时候）
        void remove(const std::string& ctag, const std::string& queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //获取消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    DLOG("没有找到队列%s的消息管理句柄", queue_name.c_str());
                    return ;
                }
                qcp = it->second;
            }
            //通过句柄完成删除
            return qcp->remove(ctag);
        }

        // 5）从指定队列获取一个消费者（轮询获取-消费者轮换消费起到负载均衡的作用）
        Consumer::ptr choose(const std::string& queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //获取消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    DLOG("没有找到队列%s的消息管理句柄", queue_name.c_str());
                    return {};
                }
                qcp = it->second;
            }
            //通过句柄完成获取消费者
            return qcp->choose();
        }

        // 6）判断队列中指定消费者是否存在
        bool empty(const std::string& queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //获取消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    DLOG("没有找到队列%s的消息管理句柄", queue_name.c_str());
                    return false;
                }
                qcp = it->second;
            }
            return qcp->empty();
        }

        // 7）判断队列中指定消费者是否存在
        bool exists(const std::string& ctag, const std::string& queue_name)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                //获取消费者管理单元
                auto it = _qconsumers.find(queue_name);
                if (it == _qconsumers.end())
                {
                    DLOG("没有找到队列%s的消息管理句柄", queue_name.c_str());
                    return false;
                }
                qcp = it->second;
            }
            //通过句柄完成删除
            return qcp->exists(ctag);
        }

        // 8）清理所有消费者
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }
        
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}